package org.apache.flink.runtime.rest.handler.job;

import java.util.Map;
import java.util.concurrent.CompletableFuture;

import javax.annotation.Nonnull;

import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.messages.webmonitor.ConsumerStatusInfo;
import org.apache.flink.runtime.resourceconsumer.ResourceConsumerGateway;
import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.RestHandlerException;
import org.apache.flink.runtime.rest.handler.resourceconsumer.AbstractResourceConsumerHandler;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.ConsumerIDPathParameter;
import org.apache.flink.runtime.rest.messages.ConsumerMessageParameters;
import org.apache.flink.runtime.rest.messages.MessageHeaders;
import org.apache.flink.runtime.rpc.retriever.GatewayRetriever;
import org.apache.flink.runtime.webmonitor.RestfulGateway;

/** Handler for requesting job status. */
/**
 * @类名: ConsumingStatusHandler
 * @说明: 消费状态请求处理器
 *
 * @author   leehom
 * @Date	 2024年3月4日 上午10:59:31
 * 修改记录：
 *
 * @see 	 
 *   
 */
public class ConsumingStatusHandler
        extends AbstractResourceConsumerHandler<
                RestfulGateway, EmptyRequestBody, ConsumerStatusInfo, ConsumerMessageParameters> {

    public ConsumingStatusHandler(
            GatewayRetriever<? extends RestfulGateway> leaderRetriever,
            Time timeout,
            Map<String, String> responseHeaders,
            MessageHeaders<EmptyRequestBody, ConsumerStatusInfo, ConsumerMessageParameters> messageHeaders,
            GatewayRetriever<ResourceConsumerGateway> resourceConsuemrGatewayRetriever) {
        super(leaderRetriever, timeout, responseHeaders, messageHeaders, resourceConsuemrGatewayRetriever);
    }

	@Override
	protected CompletableFuture<ConsumerStatusInfo> handleRequest(HandlerRequest<EmptyRequestBody> request,
			ResourceConsumerGateway gateway) throws RestHandlerException {
        ConsumerID jobId = request.getPathParameter(ConsumerIDPathParameter.class);
        return gateway.requestConsumingStatus(timeout).thenApply(ConsumerStatusInfo::new);
	}
}
